package read.niit.service


import org.apache.spark.streaming.dstream.DStream
import read.niit.bean.Reader
import read.niit.dao.Top10WithMinScoreDao
import read.niit.util.{JDBCUtil, SparkUtil}

/**
 *实时统计每个学生阅读书籍评分最低的前10本书,并带上所属书
 */

class Top10WithMinScoreService {

  val spark = SparkUtil.takeSpark()
  import spark.implicits._
  import org.apache.spark.sql.functions._

  def dataAnalysis(reader: DStream[Reader]): Unit = {
    studentMinScorewithSubjectTop10(reader)
  }

  private def studentMinScorewithSubjectTop10(reader: DStream[Reader]): Unit = {
    //(学生Id，书),评分
    val mapDS: DStream[((String, String), Int)] = reader.map(data => {
      ((data.student_id, data.book_id), data.score)
    })

    mapDS.foreachRDD(
      rdd => {

        val top10Rdd: Array[((String, String), Int)] = rdd.sortBy(_._2, true).take(10)

        println("-------实时统计每位学生阅读书籍评分最低的前10本书--------")
        top10Rdd.foreach(println)

        val top10WithMinScoreDao = new Top10WithMinScoreDao
        top10WithMinScoreDao.insertTop10Data(top10Rdd)

      }
    )

  }

}


